//
// Created by u on 23-6-28.
//

#ifndef MOSQUITTO_TEST_MQTTCLIENT_HPP
#define MOSQUITTO_TEST_MQTTCLIENT_HPP

#include <iostream>
#include <utility>
#include <vector>
#include <unordered_set>
#include <string>
#include <memory>
#include <queue>
#include <functional>
#include <condition_variable>
#include <mutex>
#include <atomic>

#include <mosquitto.h>
#include <MqttCommon.hpp>
#include <XLogger.hpp>

class MqttClient {
public:

	MqttClient(
		const std::string& clientId,
		const std::string& brokerAddress,
		int brokerPort = 1883,
		const char* name = nullptr,
		const char* pw = nullptr,
		bool cleanSession = true,
		void (*onConnect)(struct mosquitto* , void* , int) = nullptr,
		void (*onDisconnect)(struct mosquitto* , void* , int) = nullptr,
		void (*onReceiveMsg)(struct mosquitto*, void*, const struct mosquitto_message*) = nullptr)

		: clientId_(clientId), brokerAddress_(brokerAddress), brokerPort_(brokerPort), mosq_(nullptr) {

		mosquitto_lib_init();
		mosq_ = mosquitto_new(clientId_.c_str(), cleanSession, nullptr);
		if (!mosq_) {
			throw std::runtime_error("Failed to create MQTT cloudClient_ instance.");
		}

		if (onConnect)
			mosquitto_connect_callback_set(mosq_, onConnect);

		if (onDisconnect)
			mosquitto_disconnect_callback_set(mosq_, onDisconnect);

		if (onReceiveMsg)
			mosquitto_message_callback_set(mosq_, onReceiveMsg);

		if (name != nullptr && pw != nullptr)
			setUsernamePasswd(name, pw);
	}

	~MqttClient() {
		disconnect();
		mosquitto_destroy(mosq_);
		mosquitto_lib_cleanup();
	}

	bool setUsernamePasswd(const char* username, const char* pw) {
		return mosquitto_username_pw_set(mosq_, username, pw) == MOSQ_ERR_SUCCESS;
	}

	void connect(int keepAlive=60) {
		std::lock_guard<std::mutex> _lck(clientOpMutex_);
		int rc = mosquitto_connect(mosq_, brokerAddress_.c_str(), brokerPort_, keepAlive);
		if (rc != MOSQ_ERR_SUCCESS) {
			LOG_ERROR("Failed to connect to MQTT broker: ",mosquitto_strerror(rc));
		}

		mosquitto_loop_start(mosq_);
	}

	void disconnect() {
		mosquitto_disconnect(mosq_);
		mosquitto_loop_stop(mosq_, false);
	}

	bool publish(const std::string& topic, const std::string& message, int qos = 0) {
		std::cout << "topic: " << topic << " qos: " << std::to_string(qos) <<  std::endl
				  << "msg: " << message << std::endl;
		int rc = mosquitto_publish(mosq_, nullptr, topic.c_str(), message.size(), message.c_str(), qos, false);
		if (rc != MOSQ_ERR_SUCCESS) {
			LOG_ERROR("Failed to publish MQTT message: {}", mosquitto_strerror(rc));
		}
		return true;
	}

	void subscribe(const std::string& topic, int qos = 0) {
		int rc = mosquitto_subscribe(mosq_, nullptr, topic.c_str(), qos);
		if (rc != MOSQ_ERR_SUCCESS) {
			LOG_ERROR("Failed to subscribe to MQTT topic: {}", mosquitto_strerror(rc));
		}
	}

	void unsubscribe(const std::string& topic) {
		int rc = mosquitto_unsubscribe(mosq_, nullptr, topic.c_str());
		if (rc != MOSQ_ERR_SUCCESS) {
			LOG_ERROR("Failed to subscribe to MQTT topic: %s\r\n" ,mosquitto_strerror(rc));
		}
	}

private:
	std::string clientId_;
	std::string brokerAddress_;
	int brokerPort_;
	struct mosquitto* mosq_;
	std::mutex clientOpMutex_;
	static const int kMqttKeepAlive = 60;
};

#endif //MOSQUITTO_TEST_MQTTCLIENT_HPP
